package manager

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	paho "github.com/eclipse/paho.mqtt.golang"
	"github.com/zeromicro/go-zero/core/logx"
	"mqtt/internal/handler"
)

// ClientManager 管理多个MQTT服务器连接
type ClientManager struct {
	clients        map[string]paho.Client      // key: 服务器地址
	configs        map[string]Config           // 连接配置
	mu             sync.RWMutex
	msgCache       map[string][]Message        // 缓存订阅的消息（按服务器+主题）
	handlerRegistry *handler.TopicHandlerRegistry // 主题处理器注册表
}

// Config MQTT连接配置
type Config struct {
	Username string // 用户名
	Password string // 密码
	ClientID string // 客户端ID
}

// Message 缓存的MQTT消息
type Message struct {
	Topic   string
	Payload []byte
	QoS     int32
	Time    int64 // 时间戳
}

// NewClientManager 创建管理器
func NewClientManager() *ClientManager {
	return &ClientManager{
		clients:        make(map[string]paho.Client),
		configs:        make(map[string]Config),
		msgCache:       make(map[string][]Message),
		handlerRegistry: handler.NewTopicHandlerRegistry(),
	}
}

// GetHandlerRegistry 获取主题处理器注册表
func (m *ClientManager) GetHandlerRegistry() *handler.TopicHandlerRegistry {
	return m.handlerRegistry
}

// Init 初始化连接
func (m *ClientManager) Init(serverAddr string, cfg Config) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if _, exists := m.clients[serverAddr]; exists {
		return nil // 已存在连接
	}

	// 创建MQTT客户端选项
	opts := paho.NewClientOptions().AddBroker(serverAddr)
	opts.SetUsername(cfg.Username)
	opts.SetPassword(cfg.Password)
	opts.SetClientID(cfg.ClientID)
	opts.SetCleanSession(true)
	opts.SetOnConnectHandler(func(client paho.Client) {
		logx.Infof("Connected to MQTT server: %s", serverAddr)
	})

	// 创建客户端并连接
	client := paho.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		return fmt.Errorf("connect failed: %v", token.Error())
	}

	m.clients[serverAddr] = client
	m.configs[serverAddr] = cfg
	return nil
}

// Publish 发布消息
func (m *ClientManager) Publish(serverAddr, topic string, payload []byte, qos int32, retain bool) error {
	client, err := m.getClient(serverAddr)
	if err != nil {
		return err
	}

	token := client.Publish(topic, byte(qos), retain, payload)
	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("publish failed: %v", token.Error())
	}
	return nil
}

// Subscribe 订阅主题
func (m *ClientManager) Subscribe(serverAddr string, topics []string, qos int32) (int, error) {
	client, err := m.getClient(serverAddr)
	if err != nil {
		return 0, err
	}

	topicQos := make(map[string]byte)
	for _, topic := range topics {
		topicQos[topic] = byte(qos)
	}

	// 创建上下文
	ctx := context.Background()

	// 订阅消息并使用主题处理器处理
	token := client.SubscribeMultiple(topicQos, func(client paho.Client, msg paho.Message) {
		// 缓存消息
		key := fmt.Sprintf("%s|%s", serverAddr, msg.Topic())
		m.mu.Lock()
		m.msgCache[key] = append(m.msgCache[key], Message{
			Topic:   msg.Topic(),
			Payload: msg.Payload(),
			QoS:     int32(msg.Qos()),
			Time:    time.Now().Unix(),
		})
		m.mu.Unlock()

		// 获取适合该主题的处理器并处理消息
		handler := m.handlerRegistry.GetHandler(msg.Topic())
		handler.HandleMessage(ctx, client, msg)
	})

	if token.Wait() && token.Error() != nil {
		return 0, fmt.Errorf("subscribe failed: %v", token.Error())
	}
	return len(topics), nil
}

// Unsubscribe 取消订阅
func (m *ClientManager) Unsubscribe(serverAddr string, topics []string) error {
	client, err := m.getClient(serverAddr)
	if err != nil {
		return err
	}

	token := client.Unsubscribe(topics...)
	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("unsubscribe failed: %v", token.Error())
	}

	// 清理缓存
	m.mu.Lock()
	for _, topic := range topics {
		key := fmt.Sprintf("%s|%s", serverAddr, topic)
		delete(m.msgCache, key)
	}
	m.mu.Unlock()
	return nil
}

// ListSubscribedTopics 获取已订阅的主题
func (m *ClientManager) ListSubscribedTopics(serverAddr string) ([]string, error) {
	_, err := m.getClient(serverAddr)
	if err != nil {
		return nil, err
	}

	// 从客户端获取订阅的主题（paho库不直接提供，这里简化为从缓存提取）
	m.mu.RLock()
	defer m.mu.RUnlock()

	var topics []string
	prefix := serverAddr + "|"
	for key := range m.msgCache {
		if len(key) > len(prefix) && key[:len(prefix)] == prefix {
			topics = append(topics, key[len(prefix):])
		}
	}
	return topics, nil
}

// 获取客户端连接
func (m *ClientManager) getClient(serverAddr string) (paho.Client, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	client, exists := m.clients[serverAddr]
	if !exists {
		return nil, errors.New("client not initialized")
	}

	if !client.IsConnected() {
		return nil, errors.New("client not connected")
	}
	return client, nil
}

// Close 关闭所有连接
func (m *ClientManager) Close() {
	m.mu.Lock()
	defer m.mu.Unlock()

	for addr, client := range m.clients {
		if client.IsConnected() {
			client.Disconnect(250)
			logx.Infof("Disconnected from %s", addr)
		}
	}
}
